package org.anno.chunkdb;

import com.google.common.base.Preconditions;

import it.unimi.dsi.fastutil.objects.Object2IntMap;

import java.io.IOException;
import java.util.Arrays;

/**
 * Furnishes rows in ascending order of values from a given column.
 */
public class IndexScanPlan extends QueryPlan {

  private final QueryPlanSupport planSupport;
  private final byte[] inclusiveFrom;
  private final byte[] exclusiveTo;
  private boolean ascending;

  private IndexScanPlan(Metadata.Table table, int sortedColumn, QueryPlanSupport planSupport,
                        byte[] inclusiveFrom, byte[] exclusiveTo, boolean ascending) {
    super(table, sortedColumn);
    this.planSupport = planSupport;
    this.inclusiveFrom = inclusiveFrom;
    this.exclusiveTo = exclusiveTo;
    this.ascending = ascending;
  }

  public IndexScanPlan(Metadata.Table table, int sortedColumn, byte[] chunk, boolean ascending) {
    this(table, sortedColumn, new LightChunk(chunk, table), null, null, ascending);
  }

  public static IndexScanPlan from(Metadata.Table table, byte[] flatArray,
                                          Object2IntMap<String> indexByColumnName,
                                          Chunkdb.Filter filter) throws IOException {
    int column = indexByColumnName.getInt(filter.getColumn());
    Preconditions.checkArgument(column >= 0);
    byte[] rhs = getEncodedRhs(filter);
    return new IndexScanPlan(table, column, new LightChunk(flatArray, table),
                                    getInclusiveFrom(filter, rhs), getExclusiveTo(filter, rhs),
                                    true);
  }

  private static byte[] getInclusiveFrom(Chunkdb.Filter filter, byte[] rhs) {
    switch (filter.getOp()) {
      case EQ:
      case GE:
        return rhs;
      case GT:
        return Arrays.copyOf(rhs, rhs.length + 1);
      default:
        return null;
    }
  }

  private static byte[] getExclusiveTo(Chunkdb.Filter filter, byte[] rhs) {
    switch (filter.getOp()) {
      case LT:
        return rhs;
      case LE:
      case EQ:
        return Arrays.copyOf(rhs, rhs.length + 1);
      default:
        return null;
    }
  }

  @Override
  public void accept(ResponseVisitor visitor) {
    acceptForAllColumns(table, visitor);
    try {
      visitor.visitRows(
          planSupport.getOrderedRange(sortedColumn, inclusiveFrom, exclusiveTo, ascending));
    } catch (IOException e) {
      throw new IllegalArgumentException(e);
    }
  }

  private static byte[] getEncodedRhs(Chunkdb.Filter filter) {
    if (filter.hasDoubleRhs()) {
      return ValueType.encodeForOrder(Double.doubleToLongBits(filter.getDoubleRhs()));
    } else if (filter.hasLongRhs()) {
      return ValueType.encodeForOrder(filter.getLongRhs());
    }
    Preconditions.checkArgument(filter.hasStringRhs());
    return filter.getStringRhsBytes().toByteArray();
  }
}
